package com.oceansoft.mobile.econsole.modules.dcn;

import com.oceansoft.mobile.econsole.modules.dcn.dispatcher.IDispatchable;
import oracle.jdbc.OracleConnection;
import oracle.jdbc.OracleDriver;
import oracle.jdbc.OracleStatement;
import oracle.jdbc.dcn.DatabaseChangeRegistration;
import oracle.jdbc.dcn.RowChangeDescription;
import oracle.jdbc.dcn.TableChangeDescription;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Closeable;
import java.io.IOException;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.Objects;
import java.util.Properties;

/**
 * ClassName: DbChangeRegistration
 * Desc: Oracle 表数据监听器
 * Date： 2016/11/16
 * Created：shaom
 */
public class DbChangeRegistration implements Closeable {
    private final Logger logger = LoggerFactory.getLogger(getClass());

    private OracleConnection connection = null;
    private DatabaseChangeRegistration dcr = null;

    private final String dcnUrl;
    private final String dcnUser;
    private final String dcnPwd;
    private final String ntfLocalHost;

    public DbChangeRegistration(String dcnUrl, String dcnUser, String dcnPwd, String ntfLocalHost) {
        this.dcnUrl = dcnUrl;
        this.dcnUser = dcnUser;
        this.dcnPwd = dcnPwd;
        this.ntfLocalHost = ntfLocalHost;
    }

    /**
     * 添加监听
     *
     * @param dispatcher
     * @param listenSQL
     */
    public void addListener(IDispatchable dispatcher, String listenSQL) {
        try {
            connection = buildDCNConnection();
            dcr = connection.registerDatabaseChangeNotification(buildDCNConfig());
            dcr.addListener(event -> {
                if (event.getRegId() != dcr.getRegId())
                    return;
                for (TableChangeDescription tableChange : event.getTableChangeDescription()) {
                    for (RowChangeDescription rcd : tableChange.getRowChangeDescription()) {
                        //仅分发INSERT动作
                        if (rcd.getRowOperation().equals(RowChangeDescription.RowOperation.INSERT))
                            dispatcher.dispatch(tableChange.getTableName(),
                                    rcd.getRowid().stringValue(),
                                    rcd.getRowOperation());
                    }
                }
            });
            initListeningTable(listenSQL, connection);
        } catch (Exception e) {
            try {
                if (null != dcr) {
                    connection.unregisterDatabaseChangeNotification(dcr);
                    logger.error("启动Oracle数据库监听失败,取消注册监听");
                }
            } catch (SQLException ex) {
                logger.error("启动Oracle数据库监听失败,错误:{}", ex.getMessage());
                throw new RuntimeException("启动数据库监听异常", e);
            }
            logger.info("启动Oracle数据库监听失败,错误:{}", e.getMessage());
            throw new RuntimeException("启动数据库监听异常", e);
        }
    }


    /**
     * 构建DCN配置
     *
     * @return Properties
     */
    public Properties buildDCNConfig() {
        Properties dcnProperties = new Properties();
        if (null != ntfLocalHost && ntfLocalHost.trim().length() > 0) {
            dcnProperties.put(OracleConnection.NTF_LOCAL_HOST, ntfLocalHost);
        }
        dcnProperties.put(OracleConnection.CONNECTION_PROPERTY_AUTOCOMMIT, "true");
        dcnProperties.put(OracleConnection.DCN_NOTIFY_ROWIDS, "true");
        dcnProperties.put(OracleConnection.NTF_QOS_RELIABLE, "true");
        dcnProperties.setProperty(OracleConnection.DCN_BEST_EFFORT, "false");
        dcnProperties.setProperty(OracleConnection.NTF_QOS_PURGE_ON_NTFN, "false");
        dcnProperties.setProperty(OracleConnection.NTF_TIMEOUT, "0");
        return dcnProperties;
    }

    /**
     * 构建DCN数据库链接
     *
     * @return OracleConnection
     * @throws SQLException
     */
    public OracleConnection buildDCNConnection() throws SQLException {
        OracleDriver driver = new OracleDriver();
        Properties prop = new Properties();
        prop.setProperty("user", dcnUser);
        prop.setProperty("password", dcnPwd);
        return driver.connect(dcnUrl, prop).unwrap(OracleConnection.class);
    }


    /**
     * 初始化表监听
     *
     * @param listeningSQL
     * @param connection
     */
    private void initListeningTable(String listeningSQL, OracleConnection connection) {
        Statement stmt = null;
        try {
            connection.setAutoCommit(true);
            stmt = connection.createStatement();
            ((OracleStatement) stmt).setDatabaseChangeRegistration(dcr);
            ResultSet rs = stmt.executeQuery(listeningSQL);
            while (rs.next()) {
            }
            String[] tableNames = dcr.getTables();
            for (String tableName : tableNames) {
                logger.debug("DCN开始监听{}表", tableName);
            }
            rs.close();
        } catch (SQLException e) {
            throw new RuntimeException("DCN绑定监听数据表异常", e);
        } finally {
            try {
                if (stmt != null) {
                    stmt.close();
                }
            } catch (Exception e) {
                throw new RuntimeException("关闭Statement异常", e);
            }
        }
    }

    /**
     * 注销DCN监听
     */
    public void stopListening() {
        try {
            if (null != connection && !connection.isClosed() && null != dcr) {
                connection.unregisterDatabaseChangeNotification(dcr);
                connection.close();
            }
        } catch (SQLException e) {
            logger.error("关闭DCN监听失败,错误:{}", e.getMessage());
        }
    }


    @Override
    public void close() throws IOException {
        stopListening();
    }
}
